package cn.ucox.web.ms.packer;

import cn.ucox.web.ms.Bootstarpable;
import cn.ucox.web.ms.components.ftp.DexFtpClient;
import cn.ucox.web.ms.components.ftp.FtpConfig;
import cn.ucox.web.ms.config.DataExchangeConfig;
import cn.ucox.web.ms.config.DataExchangeConfig.ExchangeConfig;
import cn.ucox.web.ms.redis.RedisClient;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool2.PooledObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import static cn.ucox.web.ms.Constant.*;

/**
 * 轮询数据队列监听
 *
 * @author chenw
 * @create 2017-06-06 10:39
 * @email javacspring@gmail.com
 */
public class TransactionDataSlicer implements Bootstarpable {

    private static final Logger logger = LoggerFactory.getLogger(TransactionDataSlicer.class);

    private QueueListener listener;
    private static RedisClient redisClient;
    private FtpConfig ftpConfig;
    private DexFtpClient dexFtpClient;
    private ExchangeConfig exchangeConfig;


    static {
        redisClient = RedisClient.instance();
    }

    public TransactionDataSlicer(FtpConfig ftpConfig, DexFtpClient dexFtpClient) {
        this.ftpConfig = ftpConfig;
        this.dexFtpClient = dexFtpClient;
    }

    @Override
    public void start() {
        listener = new QueueListener();
        new Thread(listener).start();
    }

    @Override
    public void stop() {
        listener.stop();
    }

    private class QueueListener implements Runnable {

        volatile boolean runningFlag = true;
        int segmentSize;//分片数据最大值，默认为20K

        QueueListener() {
            this.runningFlag = true;
            this.segmentSize = ftpConfig.getSegmentSize();
        }

        void stop() {
            runningFlag = false;
            logger.info("数据队列监听停止");
        }

        //初始化分片数据控制变量
        long transId = 0; //事务编号
        int transDataLen = 0; //事务数据总长度
        int transItemNum = 0; //事务数据总条数
        int transDataReaded = 0;//事务数据已读取长度
        int fragNum = 0;//事务数据分片数量
        int fragIdx = 1;//数据分片编号，从1开始编号
        int fragBufferRemain = segmentSize; //分片缓冲区剩余空间
        int fragBufferOffset = 0;
        boolean hasException = false;//事务数据处理过程中是否存在异常
        //分片序列格式
        DecimalFormat fragFormat = new DecimalFormat("0000");
        byte[] fragDataBuffer = new byte[0];
        JSONObject statistics = new JSONObject();
        List<String> queueData = new ArrayList<>();

        @Override
        public void run() {
            logger.debug("启动队列监听");

            while (runningFlag) {
                long start = System.currentTimeMillis();
                String data = redisClient.brpoplpush(SYNC_QUEUE, SYNC_QUEUE_BACKUP, 0);
                if (null == data || 0 == data.length()) {
                    logger.info("出列数据为空");
                    continue;
                }
                queueData.add(data);

                //标记数据
                if (data.startsWith("##DEX")) {
                    String[] anchorInfo = data.split("`");
                    //问题数据
                    if (4 != anchorInfo.length) {
                        logger.warn("数据格式错误:{}", data);
                        continue;
                    }

                    //事务开始数据
                    if (data.startsWith("##DEX_HEAD")) {
                        //初始化事务数据标记
                        transId = Long.valueOf(anchorInfo[1]);
                        transItemNum = Integer.valueOf(anchorInfo[2]);
                        transDataLen = Integer.valueOf(anchorInfo[3]);
                        transDataReaded = 0;
                        fragNum = 0 == transDataLen % segmentSize ? transDataLen / segmentSize : transDataLen / segmentSize + 1;
                        fragIdx = 1;
                        fragDataBuffer = new byte[transDataLen < segmentSize ? transDataLen : segmentSize];
                        fragBufferRemain = fragDataBuffer.length;
                        fragBufferOffset = 0;
                        hasException = false;

                        fragFormat = new DecimalFormat(ArrayUtils.toString(new int[String.valueOf(fragNum).length()]).replaceAll("[,}{]", ""));
                        logger.debug("初始化事务数据标记:{{transId:{},transItemNum:{},transDataLen:{},fragNum:{}}", transId, transItemNum, transDataLen, fragNum);
                        continue;
                    }
                    //事务结束数据
                    if (data.startsWith("##DEX_TAIL")) {
                        //事务处理过程中存在异常
                        if (hasException) {
                            JSONObject log = new JSONObject();
                            log.put("transId", transId);
                            log.put("transItemNum", transItemNum);
                            log.put("dataLength", transDataLen);
                            log.put("errorMsg","事务数据分片,上传至FTP异常,详细见备份队列");
                            //向监控表中添加异常消息
                            redisClient.zadd(SYNC_LOG_FTP, System.currentTimeMillis(), log.toJSONString());
                            //则重置事务标识，清空输入流
                            transId = 0;
                            transItemNum = 0;
                            transDataLen = 0;
                            transDataReaded = 0;
                            fragNum = 0;
                            fragIdx = 1;
                            fragDataBuffer = new byte[0];
                            fragBufferRemain = 0;
                            fragBufferOffset = 0;
                            logger.debug("事务异常，重置事务标记");
                            //不清空备份队列,进入下一次事务数据监听
                            hasException = false;//下一次事务异常标记为无异常
                            continue;
                        }
                        //将输入流缓冲区内容写入FTP
                        if (fragBufferRemain < segmentSize) { //缓冲区有内容
                            PooledObject<FTPClient> pooledObject = null;
                            String fileName = "";
                            try {
                                pooledObject = dexFtpClient.get();
                                String fragIndex = fragFormat.format(fragIdx);
                                fileName = String.format("%s_%s_%s_%s.dex", transId, fragNum, fragIndex, transDataLen);

                                if(fragNum < fragIdx){
                                    redisClient.save("QUEUE:DATA:ERROR:"+transId, JSONObject.toJSONString(queueData));
                                    queueData.clear();
                                }else {
                                    queueData.clear();
                                }

                                boolean rt = pooledObject.getObject().appendFile(fileName, new ByteArrayInputStream(fragDataBuffer));
                                //备份分片数据至本地
                                backupSegment(fileName, new String(fragDataBuffer));
                                logger.debug("+上传分片数据至FTP{}，文件名:{},文件大小:{}byte", rt ? "成功" : "失败", fileName, fragDataBuffer.length);
                                transDataReaded = 0;
                                fragNum = 0;
                                fragIdx = 1;
                                fragDataBuffer = new byte[0];
                                fragBufferRemain = 0;
                                fragBufferOffset = 0;
                                hasException = false;
                            } catch (Exception e) {
                                backupErrorSegment(fileName, new String(fragDataBuffer));
                                logger.error("+上传分片数据至FTP错误,错误原因:{}", e.getMessage(), e);
                            } finally {
                                try {
                                    dexFtpClient.close(pooledObject);
                                } catch (Exception ex) {
                                    logger.error("释放FTP连接错误", ex);
                                }
                            }
                        }
                        //清除备份队列数据,生产环境放开此注释
                        //从列表头部(左侧)删除指定长度元素
                        String delNum = redisClient.lltrim(SYNC_QUEUE_BACKUP_BYTE, transItemNum + 2);//+2是将头尾标记也清空
                        logger.debug("清空事务{}备份事务数据成功，共计{}条,实际删除:{}", transId, transItemNum, delNum);
                        //记录日志,用于流量统计
                        statistics.put("duration", System.currentTimeMillis() - start);
                        statistics.put("transId", transId);
                        statistics.put("transItemNum", transItemNum);
                        statistics.put("dataLength", transDataLen);
                        redisClient.zadd(SYNC_MONITOR_FTP, System.currentTimeMillis(), statistics.toJSONString());
                        statistics.clear();
                        //重置事务数据标记
                        transId = 0;
                        transItemNum = 0;
                        transDataLen = 0;
                        transDataReaded = 0;
                        fragNum = 0;
                        fragIdx = 1;
                        fragDataBuffer = new byte[0];
                        fragBufferRemain = 0;
                        fragBufferOffset = 0;
                        hasException = false;
                        //记录本次事务完成
                        redisClient.hmset(SYNC_ANCHOR, new HashMap<String, String>() {{
                            put(SYNC_ANCHOR_TRANSACTION_DONE, "Y");
                        }});
                        continue;
                    }
                }

                //事务数据
                //如果此次事务状态异常，则不处理事务中后续数据，而是将数据直接转储于备份队列
                if (hasException) {
                    continue;
                }
                //业务数据
                byte[] rowDataByte = new byte[0];//行数据内容
                try {
                    rowDataByte = data.getBytes("UTF-8");
                } catch (UnsupportedEncodingException e) {
                    logger.error("内容转换字节数组错误", e);
                }
                int rowDataLength = rowDataByte.length;//行数据内容长度
                if (0 == rowDataLength) {
                    logger.error("同步数据内容为空");
                    continue;
                }
                //  行数据已读字节    行数据未读字节
                int rowDataReaded = 0, rowDataRemain = rowDataLength;
                do {
                    PooledObject<FTPClient> pooledObject = null;
                    String fileName = "";
                    try {
                        //如果记录数据比分片数据剩余空间小，则一次性全部读入
                        if (rowDataRemain <= fragBufferRemain) {
                            System.arraycopy(rowDataByte, rowDataReaded, fragDataBuffer, fragBufferOffset, rowDataRemain);
                            fragBufferOffset += rowDataRemain;
                            fragBufferRemain -= rowDataRemain;
                            rowDataReaded += rowDataRemain;
                            transDataReaded += rowDataRemain;
                            rowDataRemain = 0;
                        } else {
                            if (fragIdx > fragNum) {
                                logger.error("分片已完成，异常进入！");
                                break;
                            }
                            System.arraycopy(rowDataByte, rowDataReaded, fragDataBuffer, fragBufferOffset, fragBufferRemain);
                            rowDataRemain -= fragBufferRemain;
                            rowDataReaded += fragBufferRemain;
                            transDataReaded += fragBufferRemain;
                            fragBufferOffset = 0;//此分片数据已填充满，创建一个新的fragBuffer
                            fragBufferRemain = 0;

                            //分片已满，上传分片数据至FTP
                            //FIXME:获取实例异常时导致上传空文件
                            try {
                                pooledObject = dexFtpClient.get();
                            } catch (Exception ex) {
                                logger.error("获取FtpClient实例错误，尝试重新获取");
                                Thread.sleep(3000);
                                pooledObject = dexFtpClient.get();
                            }
                            fileName = String.format("%s_%s_%s_%s.dex", transId, fragNum, fragFormat.format(fragIdx), transDataLen);
                            boolean rt = pooledObject.getObject().appendFile(fileName, new ByteArrayInputStream(fragDataBuffer));
                            backupSegment(fileName, new String(fragDataBuffer));
                            logger.debug("-上传分片数据至FTP{}，文件名:{},文件大小:{}", rt ? "成功" : "失败", fileName, exchangeConfig.getSegmentSize());
                            fragIdx++;

                            //判断事务数据剩下未读是否大于分片最大值，若大于20K，则初始化新的fragBuffer=20K,
                            int transRemain = transDataLen - transDataReaded;
                            if (transRemain > segmentSize) {
                                fragDataBuffer = new byte[segmentSize];
                            } else {
                                fragDataBuffer = new byte[transRemain];
                            }
                            //将剩下内容填充至新缓冲区
                            fragBufferRemain = fragDataBuffer.length;
                            fragBufferOffset = 0;
                        }
                    } catch (Exception ex) {
                        hasException = true;
                        backupErrorSegment(fileName, new String(fragDataBuffer));
                        logger.error("-上传分片数据至FTP错误,错误原因:{}", ex.getMessage(), ex);
                        break;
                    } finally {
                        dexFtpClient.close(pooledObject);
                    }
                } while (rowDataRemain > 0);
            }
        }
    }

    /**
     * 备份分片数据
     *
     * @param fileName 分片文件名称
     * @param data     分片数据内容
     */
    private void backupSegment(String fileName, String data) {
        /*File backDir = new File(exchangeConfig.getSendBackup());
        if (!backDir.exists() || !backDir.isDirectory()) {
            if (!backDir.mkdirs()) {
                logger.error("创建备份目录[]失败", backDir.getPath());
            }
        }
        try {
            Path path = Paths.get(String.format("%s/%s", backDir.getPath(), fileName));
            Files.write(path, data);
        } catch (IOException e) {
            e.printStackTrace();
        }*/

        FileWriter writer = null;
        try {

            // 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
            String fullFileName = String.format("%s/%s", exchangeConfig.getSendBackup(), fileName);
            writer = new FileWriter(fullFileName);
            writer.write(data);
            logger.debug("备份FTP事务数据片断至send目录:{}成功", fileName);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (writer != null) {
                    writer.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 备份ftp上传失败分片数据
     *
     * @param fileName 分片文件名称
     * @param data     分片数据内容
     */
    private void backupErrorSegment(String fileName, String data) {
        /*File backDir = new File(exchangeConfig.getSendError());
        if (!backDir.exists() || !backDir.isDirectory()) {
            if (!backDir.mkdirs()) {
                logger.error("创建备份目录[]失败", backDir.getPath());
            }
        }
        try {
            Path path = Paths.get(String.format("%s/%s", backDir.getPath(), fileName));
            Files.write(path, data);
        } catch (IOException e) {
            e.printStackTrace();
        }*/

        FileWriter writer = null;
        try {

            // 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
            String fullFileName = String.format("%s/%s", exchangeConfig.getSendError(), fileName);
            writer = new FileWriter(fullFileName);
            writer.write(data);
            logger.debug("备份FTP事务异常数据片断至send-error目录:{}成功", fileName);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (writer != null) {
                    writer.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void setExchangeConfig(ExchangeConfig exchangeConfig) {
        this.exchangeConfig = exchangeConfig;
    }
}